#include	"DBFlushService.h"

#include	<google/protobuf/dynamic_message.h>
#include	<google/protobuf/descriptor.pb.h>
#include	<google/protobuf/compiler/importer.h>

#include	<IniFile.h>
#include	<Logger.h>
#include	<Json.h>
#include	<Utils.h>

class ImportLogCollector : public google::protobuf::compiler::MultiFileErrorCollector {
public:
	virtual void AddError(const std::string & filename, int line, int column, const std::string & message) override {
		LOG_ERR("Parsing[%s] error at(%d:%d). %s", filename.c_str(), line, column, message.c_str());
	}

	virtual void AddWarning(const std::string & filename, int line, int column, const std::string & message) override {
		LOG_ERR("Warnning[%s] at(%d:%d). %s", filename.c_str(), line, column, message.c_str());
	}
};

class Parser : public ISingleton<Parser> {
public:
	Parser() : _disk(nullptr), _importer(nullptr), _factory(), _collector() {}
	virtual ~Parser() {
		if (_importer) delete _importer;
		if (_disk) delete _disk;
	}

	bool Initialize(const std::string & path, const std::string & protos) {
		_disk = new google::protobuf::compiler::DiskSourceTree();
		if (!_disk) return false;

		_disk->MapPath("", path);

		_importer = new google::protobuf::compiler::Importer(_disk, &_collector);
		if (!_importer) return false;

		std::vector<std::string> protocols = Split(protos, ";", true);
		for (auto & file : protocols) _importer->Import(file);

		return true;
	}

	google::protobuf::Message * Create(const std::string & fulltype) {
		auto descriptor = _importer->pool()->FindMessageTypeByName(fulltype);
		if (!descriptor) {
			LOG_ERR("Failed to get message description of type : %s", fulltype.c_str());
			return nullptr;
		}

		auto creator = _factory.GetPrototype(descriptor);
		if (!creator) {
			LOG_ERR("Failed to get message Prototype of type : %s", fulltype.c_str());
			return nullptr;
		}

		return creator->New();
	}

private:
	google::protobuf::compiler::DiskSourceTree * _disk;
	google::protobuf::compiler::Importer * _importer;
	google::protobuf::DynamicMessageFactory _factory;
	ImportLogCollector _collector;
};

bool DBFlushService::Initialize(class IniFile & conf) {
	std::string redis_ip = conf.Get<std::string>("redis", "ip", "127.0.0.1");
	int redis_port = conf.Get<int>("redis", "port", 6379);
	if (!_redis.Connect(redis_ip, redis_port)) {
		LOG_ERR("Failed to connect to Redis[%s:%d]", redis_ip.c_str(), redis_port);
		return false;
	}

	_redis.SetLogHandler([](const std::string & msg) { LOG_INFO(msg.c_str()); });

	std::string host = conf.Get<std::string>("mysql", "ip", "127.0.0.1");
	int port = conf.Get<int>("mysql", "port", 3306);
	std::string user = conf.Get<std::string>("mysql", "user", "root");
	std::string pswd = conf.Get<std::string>("mysql", "pswd", "root");
	std::string db = conf.Get<std::string>("mysql", "db", "game");

	if (!_db.Open(host, port, user, pswd, db)) return false;

	std::string proto_path = conf.Get<std::string>("dbflush", "proto_path", "./settings/proto");
	std::string proto_files = conf.Get<std::string>("dbflush", "proto_files", "Core.proto");
	if (!Parser::Instance().Initialize(proto_path, proto_files)) {
		LOG_ERR("Failed to load protos [%s:%s]... ", proto_path.c_str(), proto_files.c_str());
		return false;
	}

	LOG_INFO("Application dbflush started ...");
	return true;
}

void DBFlushService::Breath() {
	__Flush();
}

void DBFlushService::Quit() {
	while (__Flush()) continue;
}

bool DBFlushService::__Flush() {
	Redis::Reply reply = _redis.Command("LPOP flush_tasks");
	if (reply.IsNil()) return false;

	do {
		if (reply.IsString()) {
			Json::Reader reader;
			Json::Value task;

			bool success = reader.parse(reply.AsString(), task);
			if (!success) break;

			success = task.isObject() && task["redis_key"].isString() && task["type"].isString() && task["is_new"].isBool();
			if (!success) break;

			std::string type = task["type"].asString();
			std::string redis_key = task["redis_key"].asString();
			bool is_new = task["is_new"].asBool();
			std::string db_key = task["db_key"].isString() ? task["db_key"].asString() : "";

			Redis::Reply data = _redis.Command("GET %s", redis_key.c_str());
			if (!data) {
				LOG_ERR("Try to flush task : %s, but data is NOT exists!", task.toStyledString().c_str());
			} else if (!data.IsString()) {
				LOG_ERR("Key[%s] in redis is not store an flush data", redis_key.c_str());
			} else {
				auto msg = Parser::Instance().Create(type);
				if (!msg) {
					LOG_ERR("Failed task : %s", task.toStyledString().c_str());
				} else if (!msg->ParseFromString(data.AsString())) {
					LOG_ERR("Key and Type mismatch. Task : %s", task.toStyledString().c_str());
				} else {
					if (is_new) {
						_db.Insert(msg->GetDescriptor()->name(), *msg);
					} else {
						_db.Update(msg->GetDescriptor()->name(), *msg, db_key);
					}
					LOG_DEBUG("FLUSH : %s", task.toStyledString().c_str());
				}

				if (msg) delete msg;
			}

			return true;
		}
	} while (false);

	LOG_ERR("Failed to flush task : %s. Format is { type = flush data full-type-name in proto, redis_key = key to find data in redis, is_new = should using INSERT instead of UPDATE ?, db_key = Primary key in db (Optional, given for update, Otherwise insert) }", reply.AsString().c_str());
	return true;
}